package com.fafu.mq.mqserver.datacenter;

import com.fafu.mq.common.BinaryTool;
import com.fafu.mq.common.MqException;
import com.fafu.mq.mqserver.core.MSGQueue;
import com.fafu.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

import static com.fafu.mq.common.BinaryTool.toBytes;

public class MessageFileManager {



    //定义一个内部类, 表示该队列的统计信息
    //优先考虑静态内部类
    public static class Stat{
        //此处直接使用public
        //简单的类就不使用private来get和set
        public int totalCount;
        public int validCount;
    }

    public void init() {
        // 暂时无用, 以备扩展
    }

    //预定消息所在的目录和文件名
    //这个方法是获取指定队列下的对应的消息文件所在的路径
    public String getQueueDir(String queueName){
        return "./data/" + queueName;
    }

    //这个方法用来获取队列的数据文件的路径

    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //这个方法用来获取队列的消息统计文件的路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName){
        //由于是文本文件, 可以使用Scanner来读取
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Stat stat = new Stat();
            Scanner in = new Scanner(inputStream);
            stat.totalCount = in.nextInt();
            stat.validCount = in.nextInt();
            return stat;
        }catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName,Stat stat){
        //使用PrintWriter来写文件
        //outputStream打开文件, 默认会将原文件清空, 相当于新的数据覆盖旧的数据
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列的消息目录
    public void createQueueFiles(String queueName) throws IOException {
        //创建队列的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在就创建这个目录
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败! baseDir =" + getQueueDir(queueName));
            }
        }

        //创建队列的数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建数据文件失败! queueDataFile =" + getQueueDataPath(queueName));
            }
        }
        //创建队列的消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建消息统计文件失败! queueStatFile = " + getQueueStatPath(queueName));
            }
        }
        //给消息统计文件设置初始值
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName,stat);
    }

    //删除队列的目录和文件
    public void destroyQueueFiles(String queueName) throws IOException {
        //创建队列的消息目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 ||!ok2 ||!ok3){
            throw new IOException("删除队列的目录和文件! baseDir = " + baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件
    //如果后续生产者给 broke server 生产消息了, 这个消息就需要记录到文件上(取决于消息是否要进行持久化)
    private boolean checkFilesExist(String queueName){
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()){
            return false;
        }
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()){
            return false;
        }
        return true;
    }


    //这个方法用来把一个新的消息放到队列的对应文件中
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 先判断对应文件目录是否存在
        if(!checkFilesExist(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName = " + queue.getName());
        }
        // 2. 把Message对象进行序列化
        byte[] messageBinary = toBytes(message);
        //通过判断传入的 queue 是否相同, 来防止多线程访问会导致线程不安全的情况
        synchronized (queue){
            // 3. 获取当前队列数据文件的长度, 用这个来计算出这个 message 对象的 offsetBeg 与 offsetEnd
            //把新的数据写进去, 此时 offsetBeg , 就是当前文件的长度 + 4
            //此时 offsetEnd 的长度就是 当前文件的长度 + 4 + 消息的长度 == offsetBeg 的长度 + 消息的长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 4. 写入数据文件, 要追加到队尾, 要标志位改成true, 否则会全部删光再添加数据
            try (OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //写入消息的长度
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息的本体
                    dataOutputStream.write(messageBinary);
                }
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(),stat);
        }
    }

    //删除队列的消息
    //此处删除是逻辑删除, 就是把硬盘中消息的属性 isValid设置成 0
    // 1. 先将文件中的这一段数据读回来, 还原成 Message 对象
    // 2. 将 isValid 改成 0
    // 3. 把上述的数据重新写回文件中
    //此处这个参数的 message 对象, 必须包含有效的 offsetBeg 和 offsetEnd
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue){
            //之前使用的 FileInputStream 只能从文件头开始读, 我们现在需要的是再文件指定位置进行访问
            //也就是随机访问, 进行读写操作
            //此处使用 RandomAccessFile
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){
                // 1. 先将文件中的这一段数据读回来, 还原成 Message 对象
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                //光标指向需要的位置
                randomAccessFile.seek(message.getOffsetBeg());
                //把数据直接填满
                randomAccessFile.read(bufferSrc);
                //转换为 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 2. 将 isValid 改成 0
                diskMessage.setIsValid((byte) 0x0);
                //此处不需要将内存中的 message 改成 0, 因为内存中马上就要被删除了, 非常好进行删除操作


                // 3. 把上述的数据重新写回文件中
                byte[] bufferDest = toBytes(diskMessage);
                //虽然上面已经 seek 过了, 但是文件的读写会导致光标发生变换, 所以要重新改变光标的位置
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }

            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0){
                stat.validCount -= 1;
            }
            writeStat(queue.getName(),stat);
        }
    }

    // 从文件中, 读取所有的消息内容, 加载到内存中(放到一个链表里)
    // 这个方法, 准备在程序启动的时候调用
    // 这里使用的是 LinkedList, 主要是为了后面进行头删操作
    // 这个方法的参数只是 queueName, 不是 MSGQueue整个对象, 这个方法不需要进行加锁, 只是需要使用名字就够了
    // 此方法是在程序启动的时候调用的, 不涉及多线程安全问题, 服务器启动并没有线程安全问题
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)){
                long currentOffset = 0;
                //一个文件中有很多消息, 必须要使用 while 来进行读取
                while (true){
                    // 通过位运算来实现的四个字节读取
                    // 1. 读取这个长度, 这里的 readInt() 可能会读取到文件的末尾(EOF)
                    // readInt() 在读取到了末尾的时候会抛出一个 EOFException 异常, 和之前很多流都不一样

                    int messageSize = dataInputStream.readInt();
                    byte[] buffer = new byte[messageSize];
                    // 2. 按照这个长度读取消息
                    int actualSize = dataInputStream.read(buffer);
                    if(actualSize != messageSize){
                        //如果不匹配, 说明文件的格式一定有错误, 格式错乱了
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName = " + queueName);
                    }
                    // 3. 把这个二进制数据反序列化成对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4. 判断是否是有效消息
                    if(message.getIsValid() == 0x0){
                        //无效数据直接跳过
                        //虽然消息无效, 但是光标也在动
                        currentOffset += 4 + messageSize;
                        continue;
                    }
                    // 5. 如果是有效数据的话, 就将 message 对象加入到链表之中, 加入前填写一下 offsetBeg 和 offsetEnd
                    //计算 offset 需要知道光标的位置, 需要自己手动计算一下位置
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += 4 + messageSize;
                    messages.add(message);
                }
            }catch (EOFException e){
                //这个 catch 并非处理异常, 而是处理正常的业务, 当文件读到末尾, 会自己抛出异常
                //不需要做什么事情
                System.out.println("[MessageFileManager] 数据恢复完成! ");
            }
        }
        return messages;
    }

    public boolean checkGc(String queueName){
        // 判定是否需要进行gc
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double)stat.validCount / (double) stat.totalCount < 0.5){
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 通过这个方法, 真正执行消息数据文件的垃圾回收操作
    // 使用复制算法来完成这个操作
    // 创建一个新的文件名字叫做 queue_data_new.txt
    // 将之前的有效消息都读出来, 放到新文件中去
    // 删除旧的文件 然后再将新的文件名称改回 queue_data.txt
    // 同时要记得改消息统计文件
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行 gc 操作是大洗牌, 不能针对该队列消息文件进行任何修改
        synchronized (queue){
            long gcBag = System.currentTimeMillis();

            // 1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()){
                throw new MqException("[MessageFileManager] gc 时发现该队列的queue_data_new.txt" +
                        "已经存在! queueName = " + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }

            // 2. 从旧的文件中读取所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            long currentOffset = 0;

            // 3. 把所有有效信息写入新文件中
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for (Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                        message.setOffsetBeg(currentOffset + 4);
                        message.setOffsetEnd(currentOffset + 4 + buffer.length);
                    }
                }
            }

            // 4. 删除旧的文件, 并把新的文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建旧的文件失败! queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 重命名失败! queueDataOldFile = " + queueDataOldFile.getAbsolutePath()
                 + "queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }

            // 5. 更新统计文件
            Stat stat = new Stat();
            stat.totalCount = messages.size();
            stat.totalCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName = " + queue.getName() +
                    ", time = " + (gcEnd - gcBag) + "ms");
        }
    }
}
